package com.atguigu.gmall.realtime.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.PatternFlatTimeoutFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

public class DwdTrafficUserJumpDetail {
    public static void main(String[] args) throws Exception {
        // TODO: 2022/5/23 基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // TODO: 2022/5/23 检查点相关设置
        /*
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());
        System.setProperty("HADOOP_USER_NAME","atguigu");
        */

        // TODO: 2022/5/23 从kakfa消费数据
        String topic = "dwd_traffic_page_log";
        String groupId = "dwd_traffic_user_jump_group";

        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer(topic, groupId));

        // TODO: 2022/5/23 数据结构转换
        SingleOutputStreamOperator<JSONObject> JsonObjDS = kafkaDS.map(JSON::parseObject);

        // TODO: 2022/5/23 插入水位线
        SingleOutputStreamOperator<JSONObject> WaterarkDS = JsonObjDS.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject jsonObj, long l) {
                        return jsonObj.getLong("ts");
                    }
                }));

        // TODO: 2022/5/23 分组
        KeyedStream<JSONObject, String> keyedDS = WaterarkDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));

        // TODO: 2022/5/23 定义模式
        Pattern<JSONObject, ?> pattern = Pattern.<JSONObject>begin("start").where(
                new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObj) {
                        String lastPageId = jsonObj.getJSONObject("page").getString("last_page_id");
                        if (lastPageId != null && lastPageId.length() > 0){
                            return false;
                        }
                        return true;
                    }
                }
        ).next("next").where(
                new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObj) {
                        String lastPageId = jsonObj.getJSONObject("page").getString("last_page_id");
                        if (lastPageId != null && lastPageId.length() > 0){
                            return false;
                        }
                        return true;
                    }
                }
        ).within(Time.seconds(10));


        // TODO: 2022/5/23 将模式引用到流上
        PatternStream<JSONObject> patternDS = CEP.pattern(keyedDS, pattern);

        // TODO: 2022/5/23 处理匹配数据
        OutputTag<JSONObject> timeoutTag = new OutputTag<JSONObject>("TimeoutTag") {
        };
        SingleOutputStreamOperator<JSONObject> processTimeDS = patternDS.flatSelect(timeoutTag, new PatternFlatTimeoutFunction<JSONObject, JSONObject>() {
                    @Override
                    public void timeout(Map<String, List<JSONObject>> map, long l, Collector<JSONObject> collector) throws Exception {
                        //处理超时数据
                        JSONObject startObj = map.get("start").get(0);
                        collector.collect(startObj);
                    }
                },
                new PatternFlatSelectFunction<JSONObject, JSONObject>() {

                    @Override
                    public void flatSelect(Map<String, List<JSONObject>> map, Collector<JSONObject> out) throws Exception {
                        List<JSONObject> startList = map.get("start");
                        for (JSONObject jsonObj : startList) {
                            out.collect(jsonObj);
                        }
                    }
                });
        // TODO: 2022/5/23 将两个流合并
        DataStream<JSONObject> unionDS
                = processTimeDS.union(processTimeDS.getSideOutput(timeoutTag));

        unionDS.print();

        // TODO: 2022/5/23 写入kafka主题中
        unionDS
                .map(jsonobj -> jsonobj.toJSONString())
                .addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_user_jump_detail"));

        env.execute();
    }
}
